package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_stream_WordCount {

  def main(args: Array[String]): Unit = {
    //初始化环境
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkStream")
    //初始化SparkStreamingContext            采集周期3秒
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    //通过监控端口创建DStream ,读取数据为一行一行  监听本机 9999端口发送的数据
    var lines:ReceiverInputDStream[String] =  ssc.socketTextStream("localhost",9999)
    //比如说 3秒内发送过来的数据是：lines = "spark hello spark hello"  ==> [spark,hello,spark,hello]
    //将每一行数据切分，形成一个个的单词
    var words:DStream[String] = lines.flatMap(_.split(" "))
    //将单词映射（转换）元组  [spark,hello,spark,hello] ==>（spark,1）(hello,1) (spark,1) (hello,1)
    var wordOne =  words.map( (_,1) )
    //将相同单词做次数统计（规约）（spark,1）(hello,1) (spark,1) (hello,1)==>(spark,2) (hello,2)
    var wordCount:DStream[(String,Int)] = wordOne.reduceByKey(_+_)
    //打印
    wordCount.print

    //1.开启采集器
    ssc.start()
    //2.等待采集器关闭
    ssc.awaitTermination()


  }

}
